package com.rn.gypsophila.rabbitmq.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * Description: 抽象的分布式事务消息生产者
 * <p>
 * 问题：当消息的生产者将消息发送出去之后，消息到底有没有正确地到达服务器呢?
 * RabbitMQ 针对这个问题，提供了两种解决方式:1）通过事务机制实现；2）通过发送方确认(publisher confirm)机制实现。
 * 相比之下，发送方确认机制最大的好处在于它是异步的
 * 基于RabbitMQ解决分布式事务，核心是最终一致性。比如电商下单减库存、外卖下单派单等场景，经过一定的时间数据最终达到一致。
 * <p>
 * 生产者发送消息到RabbitMQ有以下几种情况：
 * 1.如果没有发送到交换机的数据，会回调confirm方法（ack为false）,我们可以在发送消息时把业务参数比如订单号设置到
 * correlationData参数中，回调时把相关消息保存到异常消息表，采用定时任务兜底，并报警通知相关人员。
 * 2.如果此交换机没有匹配的队列，那么消息会丢失，所以需要结合mandatory（设置为true）,当交换机没有匹配的队列时，
 * 会回调returnedMessage方法将消息返回给生产者，我们保存消息body部分，采用定时任务兜底，并报警通知相关人员。
 * 3.交换机、队列、消息（重要的数据比如支付数据）需要持久化，避免消息在RabbitMQ重启、宕机等异常情况下造成消息丢失。
 * 4.对应消费者，需要手动应答，明确告诉RabbitMQ服务器，已经正确消费了消息，然后RabbitMQ服务器会删除已经被正确消费的消息。
 * 如果RabbitMQ没有收到应答消息（比如消费者处理超时或者网络不好），会间隔的将消息重新发给消费者消费，这时消费者需要根据
 * 消息ID或者全局唯一业务字段做好幂等处理。消费者业务处理异常，兜底方案依然是保存消息到异常消息表，定时处理并报警通知相关人员处理。
 *
 * @author 然诺
 * @date 2021/4/10
 */
@Slf4j
public abstract class AbstractTxMessageProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private Jackson2JsonMessageConverter messageConverter;

    /**
     * 发送消息
     */
    protected void send(MessageMeta metaMessage) {
        // 设置确认机制回调
        rabbitTemplate.setConfirmCallback(this);
        // 设置消息退回回调
        rabbitTemplate.setReturnCallback(this);
        rabbitTemplate.setMessageConverter(messageConverter);
        CorrelationData correlationData = new CorrelationData(metaMessage.getMessageId());
        rabbitTemplate.convertAndSend(metaMessage.getExchange(), metaMessage.getRoutingKey(), metaMessage.getPayload(),
                message -> {
                    MessageProperties messageProperties = message.getMessageProperties();
                    // 设置消息持久化
                    messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    return message;
                },
                correlationData);
    }

    /**
     * 说明：没有发送到交换机的数据，会回调confirm方法（ack为false）,我们可以在发送消息时把业务参数比如订单号设置到
     * correlationData参数中，回调时把相关消息保存到异常消息表，采用定时任务兜底，并报警通知相关人员。
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 只要消息能投入正确的交换器中，ack就为true
        if (ack) {
            log.info("消息发送确认成功!消息ID：{}", correlationData.getId());
        } else {
            log.warn("消息发送确认失败:{}", cause);
            // 处理消息没有到达交换机，数据丢失的情况
            // TODO: 这里可以采用重试机制
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        //消息到达交换机，没有路由到队列。可能原因：1.交换机没有绑定队列;2.交换机根据路由键没有匹配到队列;3.队列消息已满
        log.warn("消息被退回！messageId={} | replyCode={} | replyText={} | exchange={} | routingKey={}",
                message.getMessageProperties().getMessageId(), replyCode, replyText, exchange, routingKey);
    }
}
